package beautiful.butterfly.drds.data_exchange.core.transport.channel.memory;

import beautiful.butterfly.drds.data_exchange.core.transport.channel.RecordQueue;
import beautiful.butterfly.drds.data_exchange.core.transport.record.TerminateRecord;
import beautiful.butterfly.drds.data_exchange.core.util.FrameworkErrorCode;
import beautiful.butterfly.drds.data_exchange.core.util.container.Constants;
import common.element.Record;
import common.exception.DataExchangeException;
import common.util.Configuration;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 内存Channel的具体实现，底层其实是一个ArrayBlockingQueue
 */
public class MemoryRecordQueue extends RecordQueue
{

    private int bufferSize = 0;

    private AtomicInteger memoryByteSize = new AtomicInteger(0);

    private ArrayBlockingQueue<Record> recordQueue = null;

    private ReentrantLock lock;

    private Condition notInsufficient, notEmpty;

    public MemoryRecordQueue(final Configuration configuration)
    {
        super(configuration);
        this.recordQueue = new ArrayBlockingQueue<Record>(this.getCapacity());
        this.bufferSize = configuration.getInt(Constants.datax_core_transport_exchanger_buffersize);

        lock = new ReentrantLock();
        notInsufficient = lock.newCondition();
        notEmpty = lock.newCondition();
    }

    @Override
    public void close()
    {
        super.close();
        try
        {
            this.recordQueue.put(TerminateRecord.get());
        } catch (InterruptedException ex)
        {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void clear()
    {
        this.recordQueue.clear();
    }

    @Override
    protected void put(Record record)
    {
        try
        {
            long startTime = System.nanoTime();
            this.recordQueue.put(record);
            waitWriterTime += System.nanoTime() - startTime;
            memoryByteSize.addAndGet(record.getMemoryByteSize());
        } catch (InterruptedException ex)
        {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    protected void addAll(Collection<Record> recordCollection)
    {
        try
        {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            int recordCollectionMemoryByteSize = getRecordCollectionMemoryByteSize(recordCollection);
            while (memoryByteSize.get() + recordCollectionMemoryByteSize > this.byteCapacity || recordCollection.size() > this.recordQueue.remainingCapacity())
            {
                notInsufficient.await(200L, TimeUnit.MILLISECONDS);
            }
            this.recordQueue.addAll(recordCollection);
            waitWriterTime += System.nanoTime() - startTime;
            memoryByteSize.addAndGet(recordCollectionMemoryByteSize);
            notEmpty.signalAll();
        } catch (InterruptedException e)
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR, e);
        } finally
        {
            lock.unlock();
        }
    }

    @Override
    protected Record take()
    {
        try
        {
            long startTime = System.nanoTime();
            Record record = this.recordQueue.take();
            waitReaderTime += System.nanoTime() - startTime;
            memoryByteSize.addAndGet(-record.getMemoryByteSize());
            return record;
        } catch (InterruptedException e)
        {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    @Override
    protected void drainTo(Collection<Record> recordCollection)
    {
        assert recordCollection != null;
        recordCollection.clear();
        try
        {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (this.recordQueue.drainTo(recordCollection, bufferSize) <= 0)
            {
                notEmpty.await(200L, TimeUnit.MILLISECONDS);
            }
            waitReaderTime += System.nanoTime() - startTime;
            int recordCollectionMemoryByteSize = getRecordCollectionMemoryByteSize(recordCollection);
            memoryByteSize.addAndGet(-recordCollectionMemoryByteSize);
            notInsufficient.signalAll();
        } catch (InterruptedException e)
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR, e);
        } finally
        {
            lock.unlock();
        }
    }

    private int getRecordCollectionMemoryByteSize(Collection<Record> recordCollection)
    {
        int memoryByteSize = 0;
        for (Record record : recordCollection)
        {
            memoryByteSize += record.getMemoryByteSize();
        }
        return memoryByteSize;
    }

    @Override
    public int size()
    {
        return this.recordQueue.size();
    }

    @Override
    public boolean isEmpty()
    {
        return this.recordQueue.isEmpty();
    }

}
